#ifndef __SCHEDULER_H
#define __SCHEDULER_H

#define MAX_CLIENTS 1000
#define LISTEN 0
#define ACCEPT 1
#define CONNECT 2
#define SEND 3
#define RECEIVE 4

#include "connection.h"
#include "port.h"

#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

/***************************************/
// Global variables, shared among threads

// Semaphores for synchronization
extern pthread_mutex_t** socket_semaphore;
extern pthread_mutex_t** cq_semaphore;
extern pthread_mutex_t** portmutex;

// Read only
extern HashTable<int> connection_map;
extern HashTable<int> sock2port;
extern int* maxfd;

// Access to this one need not be synced since
// each thread will only work on its portion of
// connections
extern Heap<Exchange>* ce_master;
extern HashTable<Connection> connections;
extern Heap<Exchange>** connexchanges;

// Access to these needs to be loosely synced
extern Heap<Exchange>** dataexchanges;
extern Heap<Port>* worker_ports;
extern HashTable<Port> ports;

// Must sync access to this
extern fd_set* sockets;

// No sync needed here
extern int* condone, *conorig, total;
extern double* diff, *maxdiff;
extern double maxtime;
extern int* stalled;

/****************************************/

class Scheduler
{
  double starttime;

 public:
  Scheduler();
  ~Scheduler();
  void Run();
};

Scheduler::Scheduler()
{
}

Scheduler::~Scheduler()
{
  cout << "Scheduler exited ";
}

void Scheduler::Run()
{
  struct timeval tv;
  gettimeofday(&tv, NULL);
  starttime = tv.tv_sec+tv.tv_usec/1000000.00;
  cout << "Scheduler started at "<< starttime << " connections "<<ce_master->Size()<<endl;
  total = ce_master->Size();

  while (TRUE)
    {
      int* mfu = new int[MAXWORKERS];
      int* fu = new int[MAXWORKERS];
      int mfuthreads = 0;
      // Ensure that you cycle from different worker each time
      // Fill the queues and check if old stuff is there, then redistribute it
      for(int i=0;i<MAXWORKERS;i++)
	{
	  gettimeofday(&tv, NULL);
	  double timenow = tv.tv_sec+tv.tv_usec/1000000.00 - starttime;
	  cout <<"Scheduler: "<<timenow<<" checking worker "<<i<<" master size "<<ce_master->Size()<<endl;
	  Exchange e = connexchanges[i]->peak();
	  if (connexchanges[i]->HasMore() && e.time + DELAY < timenow)
	    {
	      // This thread is stalled, redistribute its load
	      cout << "Scheduler: Worker "<<i<<" seems stalled, exchange time "<<e.time<<endl;
	      stalled[i] = -1;
	      pthread_mutex_lock(cq_semaphore[i]);
	      while (connexchanges[i]->HasMore() && e.time < timenow + DELAY)
		{
		  e = connexchanges[i]->pop();
		  ce_master->insert(e);
		  cout<<"Scheduler: rescheduling conn "<<e.c->ID<<endl;
		  e = connexchanges[i]->peak();
		}
	      pthread_mutex_unlock(cq_semaphore[i]);
	    }
	  else
	    {
	      // Candidate for hosting rescheduled tasks
	      if (stalled[i] > 0)
		{
		  fu[mfuthreads] = stalled[i];
		  mfu[mfuthreads++] = i;
		}
	      if (connexchanges[i]->HasMore())
		{
		  cout<<"Scheduler: Worker "<<i<<" seems OK, time for exchange  "<<e.time<<" stalled "<<stalled[i]<<endl;
		}
	    }
	}
      // If there is something to reschedule
      if (ce_master->Size())
	{
	  // Sort the mfu queue
	  for (int i=0;i<mfuthreads;i++)
	    for(int j = i+1; j<mfuthreads; j++)
	      if (fu[i] < fu[j])
		{
		  int temp = fu[i], mtemp = mfu[i];
		  mfu[i]=mfu[j];
		  fu[i] = fu[j];
		  mfu[j] = mtemp;
		  fu[j] = temp;
		}
	  cout << "Scheduler: Should schedule "<<ce_master->Size()<<" exchanges"<<endl;
	  int batch = ce_master->Size();
	  for (int j=0;j<batch;j++)
	    {

	      for(int k=0;k<mfuthreads;k++)
		{
		  int i = mfu[k];
		  if (stalled[i] <= 0 || !ce_master->HasMore())
		    continue;
		  pthread_mutex_lock(cq_semaphore[i]);
		  {
		    Exchange e = ce_master->pop();
		    connexchanges[i]->insert(e);
		    cout << "Scheduler: Thread order "<<k<<" thread "<<i<<" used "<<fu[k]<<" got exchange "<<e.c->ID<<endl;
		  }
		  pthread_mutex_unlock(cq_semaphore[i]);
		}
	    }
	}
      cout<<"Scheduler: finished, master size "<<ce_master->Size()<<endl;
      for(int i=0;i<MAXWORKERS;i++)
	stalled[i] = FALSE;
      sleep(1);
    }
}

#endif
